package org.jzyan.framework.delayedqueue;

import lombok.extern.slf4j.Slf4j;
import org.jzyan.framework.delayedqueue.listener.RedisDelayedQueueListener;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 添加延迟队列工具类
 * </p>
 *
 * @author jzyan
 * @since 2021-09-15
 */
@Slf4j
@Component
public class RedisDelayedQueue {

    @Autowired
    RedissonClient redissonClient;

    /**
     * 添加队列
     *
     * @param e        DTO传输类
     * @param delay    时间数量
     * @param timeUnit 时间单位
     * @param <T>      泛型
     */
    public <T> void addQueue(T e, long delay, TimeUnit timeUnit, String queueName) {
        log.info("添加队列: {}, delay: {}, timeUnit: {}", queueName, delay, timeUnit);
        RDelayedQueue<T> delayedQueue = this.getDelayedQueue(queueName);
        delayedQueue.offer(e, delay, timeUnit);
        delayedQueue.destroy();
    }

    /**
     * 注册监听
     *
     * @param queueName
     * @param redisDelayedQueueListener
     * @param <T>
     */
    public <T> void take(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
        RBlockingQueue<T> blockingFairQueue = this.getBlockingFairQueue(queueName);
        // 防止服务重启take阻塞导致无法消费消息
        redissonClient.getDelayedQueue(blockingFairQueue);
        // 由于此线程需要常驻，可以新建线程，不用交给线程池管理
        Thread thread = new Thread(() -> {
            log.info("启动监听队列线程" + queueName);
            while (true) {
                try {
                    if (redissonClient.isShuttingDown()) {
                        return;
                    }
                    redisDelayedQueueListener.invoke(blockingFairQueue.take());
                } catch (RedissonShutdownException e) {
                    log.info("{}", e.getLocalizedMessage());
                } catch (Exception e) {
                    log.error("监听队列线程异常: ", e);
                }
            }
        });
        thread.setName(queueName);
        thread.setDaemon(true);
        thread.start();
    }

    private <T> RDelayedQueue<T> getDelayedQueue(String queueName) {
        RBlockingQueue<T> blockingFairQueue = this.getBlockingFairQueue(queueName);
        return redissonClient.getDelayedQueue(blockingFairQueue);
    }

    private <T> RBlockingQueue<T> getBlockingFairQueue(String queueName) {
        return redissonClient.getBlockingQueue(queueName);
    }

}
